/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */
package cn.ac.iie.di.handler;

import cn.ac.iie.di.datadock.rdata.exchange.client.core.REFieldType;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.REMessage;
import cn.ac.iie.di.util.GlobalParas;
import cn.ac.iie.di.util.TARUtil;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.JSONArray;
import java.io.BufferedWriter;
import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.OutputStreamWriter;
import java.text.SimpleDateFormat;
import java.util.Arrays;
import java.util.Date;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import org.apache.commons.io.FileUtils;
import org.apache.log4j.Logger;

/**
 *
 * @author Administrator
 */
public class WriteHandler implements Runnable {

    public static Logger logger = Logger.getLogger(WriteHandler.class.getName());

    private final SimpleDateFormat DATAFORMAT = new SimpleDateFormat("yyyyMMdd");

    @Override
    public void run() {
        int linenum = 0;
        long time = System.currentTimeMillis() / 1000;
        long fileNameTime = System.currentTimeMillis();
        String dateStr = DATAFORMAT.format(new Date(fileNameTime));
        String srcPackDir = GlobalParas.localPath + "pack/";
        File dirFile = new File(GlobalParas.localPath);
        if (!dirFile.exists() && !dirFile.isDirectory()) {
            dirFile.mkdirs();
        }
        File tarFile = new File(GlobalParas.packPath);
        if (!tarFile.exists() && !tarFile.isDirectory()) {
            tarFile.mkdirs();
        }
        File srcPackDirFile = new File(srcPackDir);
        if (!srcPackDirFile.exists() && !srcPackDirFile.isDirectory()) {
            srcPackDirFile.mkdirs();
        }
        String filename = dateStr + "-" + fileNameTime + "-" + UUID.randomUUID().toString().replaceAll("-", "");
        BufferedWriter writer = getBufferedWriter(GlobalParas.localPath + filename);
        while (true) {
            if (GlobalParas.oriMsgPool.isEmpty() && GlobalParas.isShouldExit) {
                logger.info("will exit the write thread : " + Thread.currentThread().getName());
                try {
                    writer.close();
                    handle(filename, srcPackDir);
                    logger.info("close final file : " + GlobalParas.localPath + filename + "  ok, will break WriteThread");
                } catch (IOException ex) {
                    logger.error("close file " + GlobalParas.localPath + filename + " failed, cause " + ex.getMessage(), ex);
                }
                break;
            }

            try {
                REMessage message = GlobalParas.oriMsgPool.poll(20, TimeUnit.MILLISECONDS);
                if (message != null) {
                    Object[] jsondata=null;
                    try {
                        int key_id = 0;
                        int msg_type_id = 0;
                        String[] colNames = message.colNames;
                        for (int k = 0; k < colNames.length; k++) {
                            if ("key".equals(colNames[k])) {
                                key_id = k;
                            } else if ("msg_type".equals(colNames[k])) {
                                msg_type_id = k;
                            }
                        }
                        jsondata = (Object[]) message.data.get(0);
                        String key = (String) jsondata[key_id];
                        String msg_type = jsondata[msg_type_id].toString();
                        logger.debug(jsondata);
                        logger.debug(key);
                        logger.debug(msg_type);
                        GlobalParas.spm.inc("dv " + key + " " + msg_type, message.data.size());

                        for (Object[] line : message.data) {
                            try {
                                JSONObject jis = new JSONObject();
                                for (int col = 0; col < message.colNames.length; col++) {
                                    jis.put(message.colNames[col], line[col]);
                                }
                                String returnJsonStr = jis.toJSONString();
                                logger.debug(returnJsonStr);
//                                String returnJsonStr = printSinaleColumnAndData(message.colNames, message.types, line);
                                if (returnJsonStr != null) {
                                    try {
                                        writer.write(returnJsonStr);
                                        writer.flush();
                                        writer.newLine();
                                    } catch (IOException ex) {
                                    }
                                    linenum++;
                                }
                                //如果文件行数达到一定，则将文件关闭，并生成相应的.ok文件
                                if (linenum >= GlobalParas.writeFileSize && linenum != 0) {
                                    try {
                                        writer.close();
                                        logger.info("close file : " + GlobalParas.localPath + filename + "  ok, will create a new file");
                                    } catch (IOException ex) {
                                    }
                                    handle(filename, srcPackDir);
                                    fileNameTime = System.currentTimeMillis();
                                    dateStr = DATAFORMAT.format(new Date(fileNameTime));
                                    filename = dateStr + "-" + fileNameTime + "-" + UUID.randomUUID().toString().replaceAll("-", "");
                                    writer = getBufferedWriter(GlobalParas.localPath + filename);
                                    linenum = 0;
                                }
                            } catch (Exception e) {
                                logger.error("parser this data error, cause " + e.getMessage(), e);
                            }
                        }//end-for
                    } catch (Exception e) {
                        logger.error("parser this package message failed ,cause " + e.getMessage(), e);
                        logger.error(e.getCause().toString());
                        logger.error(Arrays.toString(jsondata));
                    }
                }//end-if null
                //写文件的超时
                if (System.currentTimeMillis() / 1000 - time >= GlobalParas.writeTimeOut && linenum != 0) {
                    try {
                        writer.close();
                        logger.info("close file : " + GlobalParas.localPath + filename + "  ok, will create a new file");
                    } catch (IOException ex) {
                    }
                    handle(filename, srcPackDir);
                    fileNameTime = System.currentTimeMillis();
                    dateStr = DATAFORMAT.format(new Date(fileNameTime));
                    filename = dateStr + "-" + fileNameTime + "-" + UUID.randomUUID().toString().replaceAll("-", "");
                    writer = getBufferedWriter(GlobalParas.localPath + filename);
                    linenum = 0;
                    time = System.currentTimeMillis() / 1000;
                }
            } catch (Exception e) {
                logger.error("write json to file failed, cause " + e.getMessage(), e);
            }
        }
        logger.info("stop OK");
    }

    private void handle(String filename, String srcPackDir) {
        String tarPath = srcPackDir + filename + ".tar";
        TARUtil.pack(tarPath, GlobalParas.localPath + filename);
        new File(GlobalParas.localPath + filename).delete();
        int i = 0;
        while (i < 3) {
            try {
                FileUtils.copyFile(new File(tarPath), new File(GlobalParas.packPath + filename + ".tar"));
            } catch (IOException ex) {
                if (i == 2) {
                    logger.error("copy tar file to WANGZHA path failed, cause " + ex.getMessage(), ex);
                }
            }
            i++;
        }
        new File(tarPath).delete();
    }

    private BufferedWriter getBufferedWriter(String fileName) {
        BufferedWriter writer = null;
        try {
            File f = new File(fileName);
            if (!f.getParentFile().exists()) {
                f.getParentFile().mkdirs();
            }
            OutputStreamWriter write = new OutputStreamWriter(new FileOutputStream(f));
            writer = new BufferedWriter(write);
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        }
        return writer;
    }

    private String printSinaleColumnAndData(Object[] columns, Object[] types, Object[] data) {
        StringBuilder sb = new StringBuilder();
        try {
            sb.append("{");
            for (int i = 0; i < columns.length; i++) {
                if (i > 0) {
                    sb.append(",");
                }
                if (types[i].equals(REFieldType.Strings)) {
                    sb.append("\"" + columns[i] + "\":[");

                    String[] lines = (String[]) data[i];
                    if (lines == null || lines.length == 0) {
                        sb.append("]");
                    } else {
                        if (lines.length == 1) {
                            sb.append("\"" + lines[0] + "\"");
                        } else {
                            sb.append("\"" + lines[0] + "\"");
                            for (int j = 1; j < lines.length; j++) {
                                sb.append(",\"" + lines[j] + "\"");
                            }
                        }
                        sb.append("]");
                    }

                } else if (types[i].equals(REFieldType.Structs)) {
                    sb.append("\"" + columns[i] + "\":\"[");
                    Map[] structs = (Map[]) data[i];
                    if (structs == null || structs.length == 0) {
                        sb.append("]\"");
                    } else {
                        if (structs.length == 1) {
                            sb.append(structs[0].toString());
                        } else {
                            sb.append(structs[0].toString());
                            for (int j = 1; j < structs.length; j++) {
                                sb.append("," + structs[j].toString());
                            }
                        }
                        sb.append("]\"");
                    }

                } else if (types[i].equals(REFieldType.Ints) || types[i].equals(REFieldType.Longs)) {
                    sb.append("\"" + columns[i] + "\":[");

                    long[] tmpLongs = (long[]) data[i];
                    if (tmpLongs == null || tmpLongs.length == 0) {
                        sb.append("]");
                    } else {
                        if (tmpLongs.length == 0) {
                            sb.append(tmpLongs[0]);
                        } else {
                            sb.append(tmpLongs[0]);
                            for (int j = 1; j < tmpLongs.length; j++) {
                                sb.append("," + tmpLongs[j]);
                            }
                        }
                        sb.append("]");
                    }

                } else {
                    sb.append("\"" + columns[i] + "\":\"" + data[i] + "\"");
                }
            }
            sb.append("}");
        } catch (Exception e) {
            logger.error("printSinaleColumnAndData error,cause " + e.getMessage(), e);
            logger.error(Arrays.toString(types));
            logger.error(Arrays.toString(columns));
            logger.error(Arrays.toString(data));
            return null;
        }
        return sb.toString();
    }
}
